package com.zs;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

/**
 * Description:
 *使用  direct Exchange   (直接转发)
 * @author: zsq-1186
 * Date: 2017-08-18-11:38
 */
public class EmitLogDirect {


    private static final String EXCHANGE_NAME = "ex_logs_direct";
    private static final String[] SEVERITIES = { "info", "warning", "error" };

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection=factory.newConnection();

        Channel channel=connection.createChannel();
        // 声明转发器的类型
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");

        channel.confirmSelect();  //将当前channel信道设置成confirm模式

        //exchange：交换机名称
        //routingKey：路由键
        // props：消息属性字段，比如消息头部信息等等
        // body：消息主体部分

        //mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
        // 那么broker会调用basic.return方法将消息返还给生产者;当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;
        // 通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者;
        for (int i=0;i<20;i++){
            String severity = getSeverity();
            String message = severity + "_log :" + UUID.randomUUID().toString();
            channel.basicPublish(EXCHANGE_NAME,severity,true,null,message.getBytes());
            System.out.println(" [x] Sent '["+severity+"]" + message + "'");
        }

        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //此处便是执行Basic.Return之后回调的地方
                String return_message = new String(body);
                System.out.println("Basic.Return返回的结果:  "+return_message);
            }
        });

        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                //TODO
                System.out.println("ack:"+deliveryTag);
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("nack:"+deliveryTag);

            }
        });
        //该方法会等到最后一条消息得到确认或者得到nack才会结束，也就是说在waitForConfirmsOrDie处会造成当前程序的阻塞
       // channel.waitForConfirmsOrDie();
        //这里close掉了可能会收不到 returnListener的回调了
        //channel.close();
        //connection.close();
    }

    /**
     * 随机产生一种日志类型
     *
     * @return
     */
    private static String getSeverity()
    {
        Random random = new Random();
        int ranVal = random.nextInt(3);
        return SEVERITIES[ranVal];
    }
}
